package org.example.flink.windows;


import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.GlobalWindows;
import org.apache.flink.streaming.api.windowing.triggers.CountTrigger;
import org.apache.flink.streaming.api.windowing.triggers.PurgingTrigger;

public class Hello08GlobalWindow {
    public static void main(String[] args) throws Exception {
        //运行环境
        StreamExecutionEnvironment environment = StreamExecutionEnvironment.getExecutionEnvironment();
        //获取数据源-admin:3
        DataStreamSource<String> source =
                environment.socketTextStream("localhost", 19523);
        //GlobalWindow
        source.map(word -> Tuple2.of(word.split(":")[0],
                        Integer.parseInt(word.split(":")[1])), Types.TUPLE(Types.STRING, Types.INT))
                .keyBy(tuple2 -> tuple2.f0)
                .window(GlobalWindows.create())
                .trigger(PurgingTrigger.of(CountTrigger.of(5)))
                .reduce((t1, t2) -> {
                    t1.f1 = t1.f1 + t2.f1;
                    return t1;
                }).print("GlobalWindow:").setParallelism(1);
        source.map(word -> Tuple2.of(word.split(":")[0],
                        Integer.parseInt(word.split(":")[1])), Types.TUPLE(Types.STRING, Types.INT))
                .windowAll(GlobalWindows.create())
                .trigger(PurgingTrigger.of(CountTrigger.of(5)))
                .reduce((t1, t2) -> {
                    t1.f1 = t1.f1 + t2.f1;
                    return t1;
                }).print("GlobalWindow:").setParallelism(1);
        //运行环境
        environment.execute();
    }
}
